package com.express.gateway.spi.proxy.network;

import java.net.ConnectException;
import java.net.MalformedURLException;
import java.time.Instant;
import java.util.Date;
import java.util.List;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeoutException;

import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.buffer.impl.BufferImpl;
import io.vertx.core.http.*;
import io.vertx.ext.web.FileUpload;
import io.vertx.ext.web.RequestBody;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.express.gateway.common.util.StrUtil;
import com.express.gateway.common.EventBusTopicConst;
import com.express.gateway.common.GatewayAttribute;
import com.express.gateway.manage.entity.ServerURL;
import com.express.gateway.manage.entity.ServerURLInfo;
import com.express.gateway.spi.policy.ServerProxyPollingPolicy;
import com.express.gateway.manage.entity.TrackInfo;
import com.express.gateway.manage.entity.Api;
import com.express.gateway.manage.enums.LoadBalanceEnum;
import com.express.gateway.common.CommonCons;
import com.express.gateway.manage.options.ServerEntranceHttpOptions;
import com.express.gateway.spi.round.AfterHandler;

import io.netty.handler.codec.http.QueryStringEncoder;
import io.vertx.core.Future;
import io.vertx.core.MultiMap;
import io.vertx.ext.web.RoutingContext;

import static com.express.gateway.common.EventBusTopicConst.SYSTEM_PLUS_TRACK_INFO;

/**
 * VxApiRoute HTTP/HTTPS服务类型的处理器
 *
 * @author flower
 */
public class HttpProxyServiceHandlerImpl implements HttpProxyServiceHandler {
    /**
     * 日志处理
     */
    private static final Logger LOG = LogManager.getLogger(HttpProxyServiceHandlerImpl.class);
    /**
     * 当前Vertx的唯一标识
     */
    /**
     * 应用的名称
     */
    private final String appName;
    /**
     * 有否使用后置处理器
     */
    private final boolean isNext;
    /**
     * API配置
     */
    private final Api api;
    /**
     * 后端服务策略
     */
    private final ServerProxyPollingPolicy policy;
    /**
     * HTTP/HTTPS服务类型的配置
     */
    private final ServerEntranceHttpOptions servOptions;
    /**
     * HTTP客户端
     */
    private final HttpClient httpClient;

    /**
     * 初始化一个服务器
     *
     * @param appName    应用的名称
     * @param isNext     是否有后置处理器
     * @param api        API相关的配置
     * @param httpClient {@link io.vertx.core.http.HttpClient}
     * @throws MalformedURLException 错误的URI路径
     * @throws NullPointerException  少了参数
     */
    public HttpProxyServiceHandlerImpl(String appName, boolean isNext, Api api, HttpClient httpClient) throws NullPointerException, MalformedURLException {
        super();
        this.appName = appName;
        this.isNext = isNext;
        this.api = api;
        this.httpClient = httpClient;
        this.servOptions = ServerEntranceHttpOptions.fromJson(api.getServerEntrance().getBody());
        if (servOptions == null) {
            throw new NullPointerException("HTTP/HTTPS server config can't convert to object class!");
        }
        List<ServerURL> urls = servOptions.getServerUrls();
        if (urls == null || urls.isEmpty()) {
            throw new NullPointerException("the server url proxy is empty!");
        }
        if (servOptions.getBalanceType() == null) {
            servOptions.setBalanceType(LoadBalanceEnum.POLLING_AVAILABLE);
        }
        this.policy = new ServerProxyPollingPolicy(urls);
    }

    @Override
    public void handle(RoutingContext rct) {
        // 添加请求到达核心处理器的数量与当前API正在处理的数量
        // 响应用户请求结束时间,响应后检查是否有坏的服务,报告当前API已经处理完毕
        // 进入重试连接后台服务
        rct.vertx().eventBus().send(EventBusTopicConst.SYSTEM_PLUS_HTTP_API_REQUEST, null);
        rct.request().exceptionHandler(e -> {
            LOG.error("{}, create connection error", api.keyMessage(), e);
        });
        rct.response()
                .putHeader(CommonCons.SERVER, GatewayAttribute.FULL_NAME)
                .putHeader(CommonCons.CONTENT_TYPE, api.getContentType())
                .endHandler(end -> rct.vertx().eventBus().send(EventBusTopicConst.SYSTEM_MINUS_CURRENT_PROCESSING, null))
                .exceptionHandler(e -> LOG.error("{}, return result to custom error", api.keyMessage(), e));
        // 判断后台服务是否有可用连接,有可用连接进行请求,如果没有可用连接进行重试
        if (!policy.isHaveService() || policy.isHaveBadService()) {
            LOG.warn("{}, will reTry, {}", api.keyMessage(), servOptions.getRetryTime());
            retryConnBusinessServers(rct);
            return;
        }

        // 后台服务连接信息
        ServerURLInfo urlInfo;
        if (servOptions.getBalanceType() == LoadBalanceEnum.IP_HASH) {
            String ip = rct.request().remoteAddress().host();
            urlInfo = policy.getUrl(ip);
        } else {
            urlInfo = policy.getUrl();
        }
        // 获得请求服务连接
        String reqURL = urlInfo.getUrl();
        // 获取请求的Path参数并加载Path参数
        MultiMap reqPathParam = rct.get(CommonCons.KEY_PATH_PARAMS);
        if (reqPathParam.entries() != null) {
            for (Entry<String, String> entry : reqPathParam.entries()) {
                reqURL = reqURL.replace(":" + entry.getKey(), entry.getValue());
            }
        }
        // 请求服务的Header
        // MultiMap reqHeaderParam = rct.get(CommonCons.KEY_HEADER_PARAMS);
        // 请求服务的Query参数
        QueryStringEncoder reqQueryParam = rct.get(CommonCons.KEY_QUERY_PATH_PARAMS);
        // 请求服务的Body参数
        QueryStringEncoder reqBodyParam = rct.get(CommonCons.KEY_BODY_PATH_PARAMS);
        reqURL += reqQueryParam.toString();
        // 请求处理器
        RequestOptions options = new RequestOptions();
        options.setMethod(servOptions.getMethod())
                .setAbsoluteURI(reqURL)
                .setTimeout(servOptions.getTimeOut())
                .putHeader(CommonCons.USER_AGENT, GatewayAttribute.EXP_API_USER_AGENT);

        List<Entry<String, String>> entries = rct.request().headers().entries();
        for (Entry<String, String> e : entries) {
            options.getHeaders().add(e.getKey(), e.getValue());
        }
        // options.getHeaders().addAll(rct.request().headers());

        // 执行监控
        TrackInfo trackInfo = new TrackInfo(appName, api.getApiName());
        trackInfo.setRequestBufferLen(rct.get(CommonCons.KEY_CONTENT_LENGTH));
        trackInfo.setRequestTime(Instant.now());
        Future<HttpClientRequest> request = httpClient.request(options);

        // 异常处理器
        request.onFailure(e -> {
            failureProcess(rct, e, trackInfo, urlInfo);
        });
        // 1.成功与后端建立链接获得request
        request.onSuccess(req -> {
            successProcess(rct, req, reqBodyParam, trackInfo);
        });
    }


    private void retryConnBusinessServers(RoutingContext rct) {
        // 进入该地方证明没有可用链接,结束当前处理器并尝试重新尝试连接是否可用
        if (isNext) {
            rct.put(AfterHandler.PREV_IS_SUCCESS_KEY, Future.<Boolean>failedFuture(new ConnectException("Can't connection Business application servers!")));
            rct.next();
            return;
        }

        rct.response()
                .putHeader(CommonCons.DATE, StrUtil.getRfc822DateFormat(new Date()))
                .setStatusCode(api.getResult().getCantConnServerStatus())
                .end(api.getResult().getCantConnServerExample());
        LOG.warn("APP:{}, API:{}, will reTry at after {}s", appName, api.keyMessage(), servOptions.getRetryTime());
        // 进入重试连接后台服务, 有正在检查重试的服务则返回不 try
        if (policy.isCheckWaiting()) {
            return;
        }
        policy.setCheckWaiting(true);
        rct.vertx().setTimer(servOptions.getRetryTime(), testConn -> {
            LOG.debug("定时任务 -> 应用:{} -> API:{}重试连接后台服务URL...", appName, api.getApiName());
            List<ServerURLInfo> service = policy.getBadService();
            policy.setCheckWaiting(false);
            if (service == null) {
                return;
            }
            doProxyRequest(service);
        });
    }


    private void failureProcess(RoutingContext rct, Throwable e, TrackInfo trackInfo, ServerURLInfo urlInfo) {
        if (isNext) {
            rct.put(AfterHandler.PREV_IS_SUCCESS_KEY, Future.<Boolean>failedFuture(e));// 告诉后置处理器当前操作成功执行
            rct.next();
            return;
        }
        // 请求结束
        trackInfo.setEndTime(Instant.now());
        trackInfo.setSuccessful(false);
        trackInfo.setErrMsg(e.getMessage());
        trackInfo.setErrStackTrace(e.getStackTrace());
        // 如果是用户已经关闭了请求连接给予warn提示并结束执行
        rct.vertx().eventBus().send(SYSTEM_PLUS_TRACK_INFO, trackInfo.toJson());
        // 如果是连接异常返回无法连接的错误信息,其他异常返回相应的异常
        try {
            if (e instanceof ConnectException || e instanceof TimeoutException) {
                // 提交连接请求失败
                LOG.error("URL:" + urlInfo.getUrl() + ",下标:" + urlInfo.getIndex() + " 请求后端服务连接失败或者连接超时,已经提交给连接策略");
                policy.reportBadService(urlInfo.getIndex());
                rct.response()
                        .putHeader(CommonCons.DATE, StrUtil.getRfc822DateFormat(new Date()))
                        .setStatusCode(api.getResult().getCantConnServerStatus())
                        .end(api.getResult().getCantConnServerExample());
            } else {
                rct.response()
                        .putHeader(CommonCons.DATE, StrUtil.getRfc822DateFormat(new Date()))
                        .setStatusCode(api.getResult().getFailureStatus())
                        .end(api.getResult().getFailureExample());
            }
        } catch (Exception ex) {
            // 如果是用户已经关闭了请求连接给予warn提示并结束执行
            if (!ex.getMessage().contains("Response is closed")) {
                LOG.error("在请求后端服务的异常处理器中响应用户请求 异常:", ex);
            }
        }
    }


    private void successProcess(RoutingContext rct, HttpClientRequest req, QueryStringEncoder reqBodyParam, TrackInfo trackInfo) {
        LOG.debug("trans body start!");
        // 2.与后端建立链接并请求成功后得到 response
        // 返回数据时候也要透传,将从后端服务中获得的response数据直接抽取到外部response中返回
        req.response().onComplete(resp -> {
            if (resp.failed()) {
                LOG.error("APP:{}, API:{}, Path:{}, error:", api.getAppName(), api.getApiName(), api.getPath(), resp.cause());
                rct.response().end(resp.cause().getMessage());
                return;
            }
            // 设置请求响应时间
            trackInfo.setResponseTime(Instant.now());
            rct.response().putHeader(CommonCons.DATE, StrUtil.getRfc822DateFormat(new Date())).setChunked(true);
            // 透传header
            Set<String> tranHeaders = api.getResult().getTransHeaders();
            if (tranHeaders != null && !tranHeaders.isEmpty()) {
                tranHeaders.forEach(h -> {
                    String v = req.headers().get(h);
                    rct.response().putHeader(h, v);
                });
            }
            //写数据到后台
            if (req.headers().contains("UPLOAD_FILE")) {
                //写文件数据
                rct.response().setChunked(true);
                List<FileUpload> fileUploadList = rct.fileUploads();
                for (FileUpload fileUpload : fileUploadList) {
                    LOG.info("fileName:{}, uploadedFileName:{}", fileUpload.fileName(), fileUpload.uploadedFileName());
                    Vertx.currentContext().owner().fileSystem().readFile(fileUpload.uploadedFileName(), frs -> {
                        if (frs.succeeded()) {
                            Buffer result = frs.result();
                            rct.response().end(result);
                            LOG.info("buffer length:{}", result.length());
                        }
                    });
                }
            } else {
                // 写一般请求数据
                rct.response().end("upload success.");
                resp.result().body(b -> {
                    LOG.debug("remote serv return -> {}", b.result().toString());
                    rct.response().end(b.result());
                });
            }

            // 统计响应长度
            String repLen = req.headers().get(CommonCons.CONTENT_LENGTH);
            trackInfo.setResponseBufferLen(repLen == null ? 0 : StrUtil.getIntTry(repLen));
            trackInfo.setEndTime(Instant.now());
            rct.vertx().eventBus().send(SYSTEM_PLUS_TRACK_INFO, trackInfo.toJson());

            if (isNext) {
                rct.put(AfterHandler.PREV_IS_SUCCESS_KEY, Future.succeededFuture(true));// 告诉后置处理器当前操作成功执行
                rct.next();
            }
        });

        LOG.debug("send body to remote -> {}, {}, {}, {}, {}", req.getMethod(), req.getHost(), req.getPort(), req.getURI(), rct.body().asString());
//        if (!api.isPassBody()) {
//            req.end(reqBodyParam.toString().replace("?", ""));
//            return;
//        }

        // String cTpe = Optional.ofNullable(rctHeaders.get(CommonCons.CONTENT_TYPE)).orElse(rctHeaders.get(CommonCons.CONTENT_TYPE_LOWER_CASE));
        // req.putHeader(CommonCons.CONTENT_TYPE, cTpe);
        req.setChunked(true);
        // 发送数据到后端服务器
        req.end(Optional.ofNullable(rct.body()).map(RequestBody::buffer).orElse(BufferImpl.buffer()));
        LOG.debug("trans body complete!");
    }

    /**
     * 当存在坏的后台服务时重试连接后台看后台连接是否可用
     *
     * @param rct 路由上下文
     */
    public void retryConnServer(RoutingContext rct) {
        // 进入该地方证明没有可用链接,结束当前处理器并尝试重新尝试连接是否可用
        if (isNext) {
            rct.put(AfterHandler.PREV_IS_SUCCESS_KEY, Future.<Boolean>failedFuture(new ConnectException("无法连接上后台交互的服务器")));
            rct.next();
        } else {
            rct.response().putHeader(CommonCons.DATE, StrUtil.getRfc822DateFormat(new Date())).setStatusCode(api.getResult().getCantConnServerStatus()).end(api.getResult().getCantConnServerExample());
        }
        LOG.warn("应用:{} -> API: {},后台服务已不存在可用的后台服务URL,将以设定的重试时间:{}进行重试", appName, api.getApiName(), servOptions.getRetryTime());
        // 进入重试连接后台服务, 有正在检查重试的服务则返回不 try
        if (policy.isCheckWaiting()) {
            return;
        }
        policy.setCheckWaiting(true);
        rct.vertx().setTimer(servOptions.getRetryTime(), testConn -> {
            LOG.debug(String.format("应用:%s -> API:%s重试连接后台服务URL...", appName, api.getApiName()));
            List<ServerURLInfo> service = policy.getBadService();
            policy.setCheckWaiting(false);
            if (service == null) {
                return;
            }
            doProxyRequest(service);
        });
    }

    private void doProxyRequest(List<ServerURLInfo> service) {
        for (ServerURLInfo urlInfo : service) {
            RequestOptions options = new RequestOptions();
            options.setMethod(servOptions.getMethod()).setAbsoluteURI(urlInfo.getUrl()).setTimeout(servOptions.getTimeOut());
            Future<HttpClientRequest> request = httpClient.request(options);
            request.onSuccess(req -> {
                req.send().onSuccess(resp -> {
                    int code = resp.statusCode();
                    LOG.info("重试连接后台服: 务应用:{}, API:{}, HTTP CODE:{}, SUCCESS: {}", appName, api.getApiName(), code, code == 200);
                    policy.reportGreatService(urlInfo.getIndex());
                }).onFailure(ex -> {
                    LOG.error("应用:{} -> API:{}重试连接后台服务URL->失败:", appName, api.getApiName(), ex);
                });
            });
            request.onFailure(ex -> {
                LOG.error("应用:{} -> API:{}重试连接后台服务URL->失败:", appName, api.getApiName(), ex);
            });
        }
    }
}
